package org.lhh.server.core.repository;

import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.jdom.Document;
import org.jdom.input.SAXBuilder;
import org.jdom.output.XMLOutputter;
import org.lhh.server.common.utils.CollectionUtils;
import org.lhh.server.config.web.exception.EtlException;
import org.pentaho.di.cluster.ClusterSchema;
import org.pentaho.di.cluster.SlaveServer;
import org.pentaho.di.core.CheckResultInterface;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.StepLogTable;
import org.pentaho.di.core.logging.TransLogTable;
import org.pentaho.di.repository.*;
import org.pentaho.di.repository.kdr.delegates.KettleDatabaseRepositoryClusterSchemaDelegate;
import org.pentaho.di.repository.kdr.delegates.KettleDatabaseRepositoryDirectoryDelegate;
import org.pentaho.di.repository.kdr.delegates.KettleDatabaseRepositoryUserDelegate;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.metastore.stores.delegate.DelegatingMetaStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;

import java.io.InputStream;
import java.io.StringWriter;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;

/**
 * @author 辉仔
 * @since 2019-06-16
 */
public class EtlRepository implements EtlService {

    private static final Logger logger = LoggerFactory.getLogger(EtlRepository.class);

    private static final String server_db = "server_db";

    private static GenericObjectPool<EtlKettleDatabaseRepository> pool;

    private static EtlService etlService;

    //原来kettle系统资源库属于客户端，KettleDatabaseRepositoryConnectionDelegate 的类里面变量都是全局的
    //客户端只有一个用户操作当然是没有问题，但是web端存在多用户操作，由于并不是所有变量参与的方法都加了锁
    //会出现全局变量被其它用户覆盖，出现资源库数据异常报错等情况
    //所以这里考虑使用对象池来解决这个问题
    public static void init() {
        // 创建池对象工厂
        PooledObjectFactory<EtlKettleDatabaseRepository> factory = new EtlPooledObjectFactory();

        GenericObjectPoolConfig<EtlKettleDatabaseRepository> poolConfig = new GenericObjectPoolConfig<>();
        // 最大空闲数
        poolConfig.setMaxIdle(5);
        // 最小空闲数, 池中只有一个空闲对象的时候，池会在创建一个对象，并借出一个对象，从而保证池中最小空闲数为1
        poolConfig.setMinIdle(1);
        // 最大池对象总数
        poolConfig.setMaxTotal(20);
        // 逐出连接的最小空闲时间 默认1800000毫秒(30分钟)
        poolConfig.setMinEvictableIdleTimeMillis(1800000);
        // 逐出扫描的时间间隔(毫秒) 如果为负数,则不运行逐出线程, 默认-1
        poolConfig.setTimeBetweenEvictionRunsMillis(1800000 * 2L);
        // 在获取对象的时候检查有效性, 默认false
        poolConfig.setTestOnBorrow(true);
        // 在归还对象的时候检查有效性, 默认false
        poolConfig.setTestOnReturn(false);
        // 在空闲时检查有效性, 默认false
        poolConfig.setTestWhileIdle(false);
        // 最大等待时间， 默认的值为-1，表示无限等待。
        poolConfig.setMaxWaitMillis(5000);
        // 是否启用后进先出, 默认true
        poolConfig.setLifo(true);
        // 连接耗尽时是否阻塞, false报异常,ture阻塞直到超时, 默认true
        poolConfig.setBlockWhenExhausted(true);
        // 每次逐出检查时 逐出的最大数目 默认3
        poolConfig.setNumTestsPerEvictionRun(3);
        // 创建对象池
        pool = new GenericObjectPool<>(factory, poolConfig);

    }

    private static EtlService instance() {
        if (etlService == null) {
            InvocationHandler handler = new EtlServiceProxy(new EtlRepository());
            etlService = (EtlService) Proxy.newProxyInstance(handler.getClass().getClassLoader(), new Class[]{EtlService.class}, handler);
        }
        return etlService;
    }

    public static EtlKettleDatabaseRepository getRepository() {
        try {
            EtlKettleDatabaseRepository databaseRepository = pool.borrowObject();
            if (databaseRepository.isConnected()) {
                return databaseRepository;
            } else {
                databaseRepository.connect("admin", "admin");
                return databaseRepository;
            }
        } catch (Exception e) {
            throw new EtlException("<br>获取资源库连接失败</br><br>" + e.getMessage() + "</br>", e);
        }
    }

    static void closeRepository(EtlKettleDatabaseRepository EtlKettleDatabaseRepository) {
        if (EtlKettleDatabaseRepository != null)
            pool.returnObject(EtlKettleDatabaseRepository);
    }

    static DatabaseMeta databaseMetaXml() throws Exception {
        ClassPathResource classPathResource = new ClassPathResource("server_db.xml");
        InputStream inputStream = classPathResource.getInputStream();
        SAXBuilder reader = new SAXBuilder();
        Document document = reader.build(inputStream, null);
        XMLOutputter xmlOutputter = new XMLOutputter();
        StringWriter stringWriter = new StringWriter();
        xmlOutputter.output(document, stringWriter);
        String xml = stringWriter.toString();
        return new DatabaseMeta(xml);
    }

    public static TransMeta getTransMeta(String transName, ObjectId objectId) {
        return EtlRepository.instance().getTransMeta(transName, objectId, EtlRepository.getRepository());
    }

    public static void saveTrans(TransMeta transMeta) {
        EtlRepository.instance().saveTrans(transMeta, EtlRepository.getRepository());
    }


    public static void deleteTrans(String transName, ObjectId pId) {
        EtlRepository.instance().deleteTrans(transName, pId, EtlRepository.getRepository());
    }

    public static RepositoryDirectoryInterface getHomeDirectory() {
        return EtlRepository.instance().getHomeDirectory(EtlRepository.getRepository());
    }

    public static void tranMetaCheckSteps(ArrayList<CheckResultInterface> checkResultInterfaces, boolean selected, TransMeta transMeta, DelegatingMetaStore metaStore) {
        EtlRepository.instance().tranMetaCheckSteps(checkResultInterfaces, selected, transMeta, metaStore, EtlRepository.getRepository());
    }

    public static void saveClusterSchema(ClusterSchema clusterSchema) {
        EtlRepository.instance().saveClusterSchema(clusterSchema, EtlRepository.getRepository());
    }

    public static void delClusterSchema(Long id) {
        EtlRepository.instance().delClusterSchema(id, EtlRepository.getRepository());
    }

    public static List<SlaveServer> getSlaveServers() {
        return EtlRepository.instance().getSlaveServers(EtlRepository.getRepository());
    }

    public static ObjectId[] getClusterIDs(boolean includeDeleted) {
        return EtlRepository.instance().getClusterIDs(includeDeleted, EtlRepository.getRepository());
    }

    public static ClusterSchema loadClusterSchema(ObjectId objectId, List<SlaveServer> allServers) {
        return EtlRepository.instance().loadClusterSchema(objectId, allServers, EtlRepository.getRepository());
    }

    public static TransMeta loadTransformation(ObjectId transformationID) {
        return EtlRepository.instance().loadTransformation(transformationID, EtlRepository.getRepository());
    }

    public static void saveDatabaseMeta(DatabaseMeta databaseMeta) {
        EtlRepository.instance().saveDatabaseMeta(databaseMeta, EtlRepository.getRepository());
    }

    public static DatabaseMeta loadDatabaseMeta(ObjectId objectId) {
        return EtlRepository.instance().loadDatabaseMeta(objectId, EtlRepository.getRepository());
    }

    public static DatabaseMeta loadDatabaseMeta(String connection) {
        return EtlRepository.instance().loadDatabaseMeta(connection, EtlRepository.getRepository());
    }

    public static List<DatabaseMeta> readDatabases() {
        return EtlRepository.instance().readDatabases(EtlRepository.getRepository());
    }

    public static void deleteDatabaseMeta(String name) {
        EtlRepository.instance().deleteDatabaseMeta(name, EtlRepository.getRepository());
    }

    public static void saveSlaveServer(SlaveServer slaveServer) {
        EtlRepository.instance().saveSlaveServer(slaveServer, EtlRepository.getRepository());
    }

    public static void deleteSlave(ObjectId objectId) {
        EtlRepository.instance().deleteSlave(objectId, EtlRepository.getRepository());
    }

    public static RepositoryDirectoryInterface directoryTree() {
        return EtlRepository.instance().directoryTree(EtlRepository.getRepository());
    }

    public static RepositoryDirectoryInterface getDirectory(ObjectId objectId) {
        return EtlRepository.instance().getDirectory(objectId, EtlRepository.getRepository());
    }

    public static boolean existDirectory(String name, ObjectId id, ObjectId pId) {
        return EtlRepository.instance().existDirectory(name, id, pId, EtlRepository.getRepository());
    }

    public static void saveDirectory(RepositoryDirectory repositoryDirectory) {
        EtlRepository.instance().saveDirectory(repositoryDirectory, EtlRepository.getRepository());
    }

    public static void renameDirectory(String name, ObjectId id, ObjectId pId) {
        EtlRepository.instance().renameDirectory(name, id, pId, EtlRepository.getRepository());
    }

    public static void removeDirectory(ObjectId id) {
        EtlRepository.instance().removeDirectory(id, EtlRepository.getRepository());
    }

    public static void moveDirectory(ObjectId id, ObjectId pId) {
        EtlRepository.instance().moveDirectory(id, pId, EtlRepository.getRepository());
    }

    public static List<TransMeta> getTransMetas(ObjectId objectId) {
        return EtlRepository.instance().getTransMetas(objectId, EtlRepository.getRepository());
    }

    public static List<RepositoryElementMetaInterface> getTransformationObjects(LongObjectId objectId) {
        return EtlRepository.instance().getTransformationObjects(objectId, EtlRepository.getRepository());
    }

    public static boolean existTransMeta(String name, ObjectId id, LongObjectId pId) {
        return EtlRepository.instance().existTransMeta(name, id, pId, EtlRepository.getRepository());
    }

    public static UserInfo getUserInfo(String username) {
         return EtlRepository.instance().getUserInfo(username, EtlRepository.getRepository());
    }

    @Override
    public RepositoryDirectoryInterface getHomeDirectory(EtlKettleDatabaseRepository repository) {
        try {
            return repository.getUserHomeDirectory();
        } catch (Exception e) {
            throw new EtlException("获取资源目录失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void tranMetaCheckSteps(ArrayList<CheckResultInterface> checkResultInterfaces, boolean selected, TransMeta transMeta, DelegatingMetaStore metaStore, EtlKettleDatabaseRepository repository) {
        transMeta.checkSteps(checkResultInterfaces, selected, null, transMeta, repository, metaStore);
    }

    @Override
    public void saveClusterSchema(ClusterSchema clusterSchema, EtlKettleDatabaseRepository repository) {
        try {
            repository.save(clusterSchema, null, null);
        } catch (KettleException e) {
            throw new EtlException("保存集群失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void delClusterSchema(long id, EtlKettleDatabaseRepository repository) {
        try {
            repository.deleteClusterSchema(new LongObjectId(id));
        } catch (KettleException e) {
            throw new EtlException("删除集群失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public List<SlaveServer> getSlaveServers(EtlKettleDatabaseRepository repository) {
        try {
            return repository.getSlaveServers();
        } catch (KettleException e) {
            throw new EtlException("获取子服务器失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public ObjectId[] getClusterIDs(boolean includeDeleted, EtlKettleDatabaseRepository repository) {
        try {
            return repository.getClusterIDs(includeDeleted);
        } catch (KettleException e) {
            throw new EtlException("获取集群失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public ClusterSchema loadClusterSchema(ObjectId objectId, List<SlaveServer> allServers, EtlKettleDatabaseRepository repository) {
        try {
            KettleDatabaseRepositoryClusterSchemaDelegate clusterSchemaDelegate = repository.clusterSchemaDelegate;
            return clusterSchemaDelegate.loadClusterSchema(objectId, allServers);
        } catch (KettleException e) {
            throw new EtlException("获取集群失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public TransMeta loadTransformation(ObjectId transformationID, EtlKettleDatabaseRepository repository) {
        try {
            return repository.loadTransformation(transformationID, null);
        } catch (KettleException e) {
            throw new EtlException("获取转换失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void deleteTrans(String transName, ObjectId pId, EtlKettleDatabaseRepository repository) {
        try {
            ObjectId transformationID = repository.getTransformationID(transName, getDirectory(pId));
            if (transformationID != null)
                repository.deleteTransformation(transformationID);
        } catch (KettleException e) {
            throw new EtlException("删除转换失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void saveDatabaseMeta(DatabaseMeta databaseMeta, EtlKettleDatabaseRepository repository) {
        try {
            repository.save(databaseMeta, null);
        } catch (KettleException e) {
            throw new EtlException("保存数据连接失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public DatabaseMeta loadDatabaseMeta(ObjectId objectId, EtlKettleDatabaseRepository repository) {
        try {
            return repository.loadDatabaseMeta(objectId, null);
        } catch (KettleException e) {
            throw new EtlException("获取数据连接失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public DatabaseMeta loadDatabaseMeta(String connection, EtlKettleDatabaseRepository repository) {
        try {
            ObjectId databaseID = repository.getDatabaseID(connection);
            return repository.loadDatabaseMeta(databaseID, null);
        } catch (KettleException e) {
            throw new EtlException("获取数据连接失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public List<DatabaseMeta> readDatabases(EtlKettleDatabaseRepository repository) {
        try {
            return repository.readDatabases();
        } catch (KettleException e) {
            throw new EtlException("获取数据连接失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void deleteDatabaseMeta(String name, EtlKettleDatabaseRepository repository) {
        try {
            repository.deleteDatabaseMeta(name);
        } catch (KettleException e) {
            throw new EtlException("删除数据连接失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void saveSlaveServer(SlaveServer slaveServer, EtlKettleDatabaseRepository repository) {
        try {
            repository.save(slaveServer, null, null);
        } catch (KettleException e) {
            throw new EtlException("保存子服务器失败</br>" + e.getMessage(), e);
        }

    }

    @Override
    public void deleteSlave(ObjectId objectId, EtlKettleDatabaseRepository repository) {
        try {
            repository.deleteSlave(objectId);
            repository.commit();
        } catch (KettleException e) {
            throw new EtlException("删除子服务器失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public RepositoryDirectoryInterface directoryTree(EtlKettleDatabaseRepository repository) {
        RepositoryDirectory userHomeDirectory = (RepositoryDirectory) getHomeDirectory();
        RepositoryDirectoryInterface repositoryDirectoryInterface;
        try {
            repositoryDirectoryInterface = repository.loadRepositoryDirectoryTree(userHomeDirectory);
            System.out.println(repositoryDirectoryInterface);
        } catch (KettleException e) {
            throw new EtlException("获取资源目录失败</br>" + e.getMessage(), e);
        }
        return repositoryDirectoryInterface;
    }

    @Override
    public RepositoryDirectoryInterface getDirectory(ObjectId objectId, EtlKettleDatabaseRepository repository) {
        KettleDatabaseRepositoryDirectoryDelegate directoryDelegate = repository.directoryDelegate;
        RepositoryDirectory repositoryDirectory = new RepositoryDirectory();
        try {
            directoryDelegate.loadRepositoryDirectory(repositoryDirectory, objectId);
        } catch (KettleException e) {
            throw new EtlException("获取资源目录失败</br>" + e.getMessage(), e);
        }
        return repositoryDirectory;
    }


    @Override
    public boolean existDirectory(String name, ObjectId id, ObjectId pId, EtlKettleDatabaseRepository repository) {
        try {
            String[] directoryNames = repository.getDirectoryNames(pId);
            List<String> list = new ArrayList<>(Arrays.asList(directoryNames));
            if (id == null)
                return list.contains(name);
            else {
                RepositoryDirectoryInterface directory = getDirectory(id);
                list.remove(directory.getName());
                return list.contains(name);
            }
        } catch (KettleException e) {
            throw new EtlException("获取资源目录失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void saveDirectory(RepositoryDirectory repositoryDirectory, EtlKettleDatabaseRepository repository) {
        try {
            repository.saveRepositoryDirectory(repositoryDirectory);
        } catch (KettleException e) {
            throw new EtlException("保存资源目录失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void renameDirectory(String name, ObjectId id, ObjectId pId, EtlKettleDatabaseRepository repository) {
        KettleDatabaseRepositoryDirectoryDelegate directoryDelegate = repository.directoryDelegate;
        try {
            directoryDelegate.renameDirectory(id, pId, name);
        } catch (KettleException e) {
            //ignored
        }
    }

    @Override
    public void removeDirectory(ObjectId id, EtlKettleDatabaseRepository repository) {
        RepositoryDirectoryInterface directory = getDirectory(id);
        KettleDatabaseRepositoryDirectoryDelegate directoryDelegate = repository.directoryDelegate;
        try {
            directoryDelegate.deleteDirectory(directory);
        } catch (KettleException e) {
            throw new EtlException("删除资源目录失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public void moveDirectory(ObjectId id, ObjectId pId, EtlKettleDatabaseRepository repository) {
        RepositoryDirectoryInterface directory = getDirectory(id);
        RepositoryDirectoryInterface parentDirectory = getDirectory(pId);
        directory.setParent(parentDirectory);
        saveDirectory((RepositoryDirectory) directory);
    }

    @Override
    public List<RepositoryElementMetaInterface> getTransformationObjects(LongObjectId objectId, EtlKettleDatabaseRepository repository) {
        try {
            return repository.getTransformationObjects(objectId, false);
        } catch (KettleException e) {
            throw new EtlException("获取转换失败</br>" + e.getMessage(), e);
        }
    }

    @Override
    public boolean existTransMeta(String name, ObjectId id, ObjectId pId, EtlKettleDatabaseRepository repository) {
        List<RepositoryElementMetaInterface> transformationObjects = getTransformationObjects(new LongObjectId(pId));
        boolean exist = false;
        if (!CollectionUtils.isEmpty(transformationObjects)) {
            for (RepositoryElementMetaInterface repositoryElementMetaInterface : transformationObjects) {
                if (id == null)
                    exist = Objects.equals(repositoryElementMetaInterface.getDescription(), name);
                else {
                    TransMeta transMeta = loadTransformation(id);
                    exist = !Objects.equals(transMeta.getDescription(), name) && Objects.equals(repositoryElementMetaInterface.getDescription(), name);
                }
                if (exist)
                    break;
            }
        }
        return exist;
    }


    @Override
    public UserInfo getUserInfo(String username, EtlKettleDatabaseRepository repository) {
        KettleDatabaseRepositoryUserDelegate userDelegate = repository.userDelegate;
        try {
            IUser iUser = userDelegate.loadUserInfo(new UserInfo(), username);
            return (UserInfo) iUser;
        } catch (KettleException e) {
            return null;
        }
    }

    @Override
    public TransMeta getTransMeta(String transName, ObjectId objectId, EtlKettleDatabaseRepository repository) {
        try {
            return repository.loadTransformation(repository.getTransformationID(transName, getDirectory(objectId)), null);
        } catch (Exception e) {
            throw new EtlException("获取转换失败</br>" + e.getMessage(), e);
        }
    }


    @Override
    public List<TransMeta> getTransMetas(ObjectId objectId, EtlKettleDatabaseRepository repository) {
        List<TransMeta> transMetas = new ArrayList<>();
        try {
            String[] transformationNames = repository.getTransformationNames(objectId, false);
            if (!CollectionUtils.isEmpty(transformationNames)) {
                for (String transformationName : transformationNames) {
                    ObjectId transformationID = repository.getTransformationID(transformationName, getDirectory(objectId));
                    if (transformationID != null) {
                        TransMeta transMeta = repository.loadTransformation(transformationID, null);
                        transMetas.add(transMeta);
                    }
                }
            }
        } catch (KettleException e) {
            throw new EtlException("获取转换失败</br>" + e.getMessage(), e);
        }
        return transMetas;
    }

    @Override
    public void saveTrans(TransMeta transMeta, EtlKettleDatabaseRepository repository) {
        try {
            TransLogTable transLogTable = transMeta.getTransLogTable();
            transLogTable.setConnectionName(server_db);
            transLogTable.setTableName("br_trans_log");
            transLogTable.setLogInterval("1");
            StepLogTable stepLogTable = transMeta.getStepLogTable();
            stepLogTable.setConnectionName(server_db);
            stepLogTable.setTableName("br_step_log");
            repository.save(transMeta, null, null, false);
        } catch (KettleException e) {
            throw new EtlException("保存转换失败</br>" + e.getMessage(), e);
        }
    }
}
